#include "transContentStreamMemoryIn.hpp"

/////////////////////////////////////////////////////////////////////////////////////////////////////////////
/// receive data from remote port
///		set parameters, and receive, default is block style
///
///	parameters:
///		flag: the receive flag, now just support 'RECEIVE_FLAG_NONBLOCK'
///		buffer: the buffer to store data
///		bufferLen: the buffer size
///		timeout: wait receive timeout, default is int_max
///		rscb: callback method
///
///	return:
///		E_TRANC_TIMEOUT: wait timeout
///		E_TRANC_NONBLOCK: use async style, not block the caller
///		datalen: received data length
/////////////////////////////////////////////////////////////////////////////////////////////////////////////
int TransStreamInMemory::receive(int flag, void* buffer, int bufferLen, int timeout, recvSyncCallback rscb){
	timespec ts;
	if(flag != 0 || flag != RECEIVE_FLAG_NONBLOCK){
		logd_error("received, parameter error, will use block style");
		flag = 0;
	}
	this->timeout = timeout;
	this->receiveCallback = rscb;

	//check avaliable data
	if(mAvaliableCount == 0 && flag == RECEIVE_FLAG_NONBLOCK){
		logd_error("received, no data avaliable to get, and set noblock flag");
		return E_TRANC_NONBLOCK;
	}
	if(rscb != NULL){
		logd_info("received, set callback, will use callback to get data");
		callbackRecieveData();
		return E_TRANC_NONBLOCK;
	}
	/// callback is null, and blcok style or avaliable data
	if(mAvaliableCount > 0){
		return getData(buffer, bufferLen);
	}

	/// now no data and block
	ts = TransContentAssis::TimeTp2Ts(this->timeout);
	pthread_mutex_lock(&waitMutex);
	if(pthread_cond_timedwait(&waitCond, &waitMutex, &ts) == ETIMEDOUT){
		logd_error("received waited timeout!");
		return E_TRANC_TIMEOUT:;
	}
	pthread_mutex_unlock(&waitMutex);
	///now data avaliable
	logd_info("receive, now data avaliale!!");
	return getData(buffer, bufferLen);
}


/////////////////////////////////////////////////////////////////////////////////////////////////////
/// get data from buffer
///		copy avaliable data to buffer
///
/// parameters:
///		buffer: the user supplied data to store data
///		len: the buffer length
///
///	return:
///		readBytes: the ready bytes
///		E_TRANC_NO_DATA_AVALIABLE: no data avaliable
///		E_TRANC_ParamError: the user supplied parameter error
/////////////////////////////////////////////////////////////////////////////////////////////////////
int getData(char* buffer, int len){
	int leftBytes = len;
	int readCount = 0;

	/// check whether have data to read 
	if(mRecordLeftBytes == 0){
		logd_info("get data, no avaliable data to read");
		return E_TRANC_NO_DATA_AVALIABLE;
	}

	while(leftBytes > 0){
		logd_info("get data, the user supplued buffer still left %d bytes to read", leftBytes);
		if(leftBytes >  mRecordLeftBytes){
			logd_info("user supplied buffer left byte:%d larger than avaliable data reocrd left bytes:%d, so read all avaliable data in avaliable record once", leftBytes, mRecordLeftBytes);
			memcpy( buffer + readCount, avaliableData + mRecordInnerOffset, mRecordLeftBytes);
			readCount += mRecordLeftBytes;
			leftBytes -= mRecordLeftBytes;
			mRecordLeftBytes = 0;
			recordBuffer[mAvaliableIndex]  = -1;

			logd_info("get data, read current avaliable data over, check more avaliable data reocrd");
			mRecordInnerOffset = getStreamPacketHeaderLength();
			--mAvaliableCount;
			mAvaliableIndex = (mAvaliableIndex + 1) % mCapacity;

			if(mAvaliableCount > 0){
				logd_info("get data, there is still an avaliable data record");
				avaliableData = TransContentMemory::getInstance()->indexToBuffer(recordBuffer[mAvaliableIndex]);
				recordLeftBytes = getStreamPacketDataLength(avaliableData);
				logd_info("get data, reset the avaliable data index and left bytes to read");
				continue;
			}else{
				logd_info("get data, there is no data to read, reture read byte: %d", readCount);
				return readCount;
			}
		}else{
			logd_info("get data, user supplyed buffer size:%d can not store all left bytes: %d", leftBytes, mRecordLeftBytes);
			memcpy(buffer+readCount, avaliableData + mRecordInnerOffset, leftBytes);
			readCount += leftBytes;
			mRecordLeftBytes -= leftBytes;
			mRecordInnerOffset += leftBytes;
			logd_info("get data, the user supplied buffer full, total received data size:%d", readCount);
			return readCount;
		}
	}

	logd_error("get data, user supplied buffer size%d <= 0", leftBytes);
	return E_TRANC_ParamError;
}

///////////////////////////////////////////////////////////////////
/// now only support data in
///
/// parameters:
///		content: the content buffer
///		index: the content buffer index in memory pool
///
/// return:
///		RESPONSE_RESULT_NOACTION: the port manager should do nothing
///		RESPONSE_RESULT_RELEASEBUFFER: the port manager should release the buffer
///		RESPONSE_RESULT_POST_CONITNUE: the content should go on post
/////////////////////////////////////////////////////////////////////////	
int response(char* content, int index){
	int mainCode = content[0];
	int subCode = content[1];

	if(mainCode != message_type_out_data_stream){
		logd_error("stream in response, the port instance not deal with main code: %d", mainCode);
		return RESPONSE_RESULT_POST_CONITNUE;
	}
	switch(subCode){
		case message_type_data_stream_sub_dataIn:
			logd_error("stream in response, write data to pool");
			return writeData(content, index);
		default:
			logd_error("stream in response, not supprot the subCode: %d", subCode);
			return RESPONSE_RESULT_POST_CONITNUE;
	}
	logd_error("stream in response, should not go here");
	return RESPONSE_RESULT_POST_CONITNUE;
}

//////////////////////////////////////////////////////////////////////////////////
/// send ack data to remote to ack data
///
/// return:
///		E_TRANC_SUCESS
/////////////////////////////////////////////////////////////////////////////////
int ackPacket(){
	char buffer[28]={0};
	debug_enter();

	logd_info("start to construct packet");
	*(short*(buffer)) = (short)22;
	buffer[2] = (char)message_type_out_data_stream;
	*(NodeAddr*(buffer+3)) = target;

	buffer[11] = CLIENT_STACK_FLAG_BITS_FIXEDPATH; //flag
	*(short*(buffer+12))= TransContentAssis::SHORTL2N(10); //10
	buffer[14] = (char)message_type_out_data_stream;
	buffer[15] = (char)message_type_data_stream_sub_dataACK;
	*(int*(buffer+16)) = TransContentAssis::INTL2N(remotePort);//remote port
	*(int*(buffer+20)) = TransContentAssis::INTL2N(mStreamValueStart); //stream value

	/// sned to stack
	TransContentDispatcher::sendDataDataToStack(buffer, 24);
	return E_TRANC_SUCESS;
}

//////////////////////////////////////////////////////////////////////////
/// insert the index to buffer stack
///		store the buffer content to dataPool
///
/// parameters:
///		buffer: the data content
///		index: the buffer index of the memory pool
///
/// return:
///		RESPONSE_RESULT_RELEASEBUFFER: release the buffer after the response finished
///		RESPONSE_RESULT_NOACTION: do nothing, do not release the buffer
///////////////////////////////////////////////////////////////////////////
int writeData(char* buffer, int index){
	int __fragmentId = 0;
	int __insertIndex = -1;

	/// check buffer full
	if(mAvaliableCount == mCapacity){
		logd_error("write data to stream buffer, the avaliable count reached to capaticy");
		return RESPONSE_RESULT_RELEASEBUFFER;
	}	
	
	/// get offset value
	__fragmentId = getStreamPacketStreamValue(buffer);
	__insertIndex = __fragmentId - mStreamValueStart;
	logd_info("write data to stream buffer, get fragment id: %d, and the offset to mstreamStartValue:%d", __fragmentId, __insertIndex);

	/// check offset value <0 or > leftcount
	if(__insertIndex <0 || __insertIndex >= leftcount){
		logd_error("write data to stream buffer, the content insert offset out of range: <0 || >= %d", leftcount);
		return RESPONSE_RESULT_RELEASEBUFFER;
	}

	/// get insert index
	__insertIndex = (__insertIndex + mInsertIndex) % mCapacity;

	/// check the index value is -1
	if(recordBuffer[__insertIndex] != -1){
		logd_error("write data to stream buffer, connect layer error, the already received data received again");
		return RESPONSE_RESULT_RELEASEBUFFER;
	}

	/// write data in
	recordBuffer[__insertIndex] = index;

	/// check current offset is the first avaliable index
	if(__insertIndex != mInsertIndex){
		logd_info("the __insertIndex=%d != mInsertIndex=%d, so there must be some blank buffer between them");
		return RESPONSE_RESULT_NOACTION;
	}

	/// reset avaliable count and left count
	while(recordBuffer[__insertIndex] != -1 && mAvaliableCount < mCapacity){
		__insertIndex = (__insertIndex + 1)% mCapacity;
		--mLeftCount;
		++mAvaliableCount;
		++mStreamValueStart;
	}

	logd_info("write data to stream buffer, now have %d buffer already!", mAvaliableCount);
	mInsertIndex =  __insertIndex;
	__insertIndex = (__insertIndex - 1 + mCapacity)% mCapacity;
	buffer = TransContentMemory::getInstance()->indexToBuffer(recordBuffer[__insertIndex]);
	mStreamValueStart = getStreamPacketStreamValue(buffer);

	/// set left bytes
	if(mRecordLeftBytes==0){
		avaliableData = TransContentMemory::getInstance()->indexToBuffer(recordBuffer[mAvaliableIndex]);
		mRecordLeftBytes = getStreamPacketDataLength(avaliableData);
		dataAvaliabkeSignal();
	}

	/// construct ack packet
	ackPacket();

	return RESPONSE_RESULT_NOACTION;
}

void TransStreamInMemory::callbackRecieveData(){
	static int firstCall=1;
	if(firstCall == 1){
		firstCall = 0;
	}
	
	if(firstCall){
		pthread_create(&calbackThreadId, NULL, callbackRoute, this);
		pthread_detach(calbackThreadId);
	}
}

void TransStreamInMemory::dataAvaliabkeSignal(){
	debug_enter();
	pthread_mutex_lock(&waitMutex);
	pthread_cond_signal(&waitCond);
	pthread_mutex_unlock(&waitMutex);
	debug_leave();
}

void* TransStreamInMemory::callbackRoute(void* p){
	(TransStreamInMemory*(p))->callbackCore();
	return NULL;
}

void TransStreamInMemory::callbackCore(){
	while(1){
		pthread_mutex_lock(&waitMutex);
		pthread_cond_wait(&waitCond, &waitMutex);
		pthread_mutex_unlock(&waitMutex);

		receiveCallback(vid);
	}
}

